package com.zhang.b_async_confirm;

import com.rabbitmq.client.*;
import org.junit.Test;

import java.io.IOException;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;


public class A_producer {

    @Test
    public void test1() throws Exception {
            //1.创建Connection
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.200.114");
            factory.setPort(5672);
            factory.setVirtualHost("/");
            factory.setUsername("guest");
            factory.setPassword("guest");
            Connection connection = factory.newConnection();

            //2.创建channel
            Channel channel = connection.createChannel();

            //3.创建交换机(这里使用的是默认交换机 AMQP_Default)
            channel.exchangeDeclare("e1", BuiltinExchangeType.DIRECT,false,false,false,null);

            //4.创建队列
            channel.queueDeclare("q1",true,false,false,null);
            channel.queueDeclare("q2",true,false,false,null);

            //5.交换机与队列绑定(指定路由key)
            channel.queueBind("q1","e1","aaa");
            channel.queueBind("q2","e1","bbb");



            //(1)未确认集合(里面装的是未确认消息的id(deliveryTag) )
            final SortedSet<Long> confirmSet  = Collections.synchronizedNavigableSet(new TreeSet<Long>());

            //(2)开启确认模式
            channel.confirmSelect();

            //(3)信道监听是否确认
            channel.addConfirmListener(new ConfirmListener() {

                //已确认
                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    if(multiple){
                        System.out.println("确认队列收到多条: "+deliveryTag);
                        confirmSet.headSet(deliveryTag+1L).clear();//清除当前序列号+1之前的集合元素
                        System.out.println("集合情况:"+confirmSet);
                    }else {
                        System.out.println("确认队列收到单条"+deliveryTag);
                        confirmSet.remove(deliveryTag);//移除当前序列号元素
                    }
                }

                //未确认
                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    if(multiple){
                        System.out.println("多条未确认:"+deliveryTag);
                    }else {
                        System.out.println("单条未确认"+deliveryTag);
                    }
                }
            });


            //6.发送消息
            for (int i = 1; i <= 50; i++) {
                //(4)获取发送消息的序列号(deliveryTag)(唯一id)
                long seqNo = channel.getNextPublishSeqNo();
                System.out.println("消息"+i+"的序列号: "+seqNo);

                String message = "hello boy"+i;
                channel.basicPublish("e1","aaa",null,message.getBytes());

                //(5)将消息的唯一id添加至 维护消息id的集合中
                confirmSet.add(seqNo);
            }

    }
}


















